package com.winit.core.mq.queue.impl;

import com.winit.core.mq.listener.QueueListener;
import com.winit.core.mq.message.Message;
import com.winit.core.mq.queue.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by xiangyu.liang on 2016/8/15.
 */
public class AsyncMessageQueue implements MessageQueue{

    private static Logger logger = LoggerFactory.getLogger(AsyncMessageQueue.class);
    private String name;
    private BlockingQueue<Message> queue;
    private List<QueueListener> listeners;
    private Long expireTime;
    private boolean started=false;
    private static Long DEFAULT_TIME=86400000L;
    AsyncMessageQueue.Worker worker = new AsyncMessageQueue.Worker();

    @Override
    public boolean hasListener() {
        return this.listeners.size()>0;
    }

    @Override
    public void destroy() {
        this.stop();
    }

    public AsyncMessageQueue(String name){
        this.name=name;
        this.queue=new LinkedBlockingQueue<>(500);
        this.listeners =new ArrayList<>();
        this.expireTime=System.currentTimeMillis()+DEFAULT_TIME;
        this.start();
    }
    @Override
    public boolean enqueue(Message message){
        return this.queue.offer(message);
    }
    protected boolean isStarted(){
        return this.started;
    }
    public void start(){
        this.started=true;
        this.worker.start();
    }
    public void stop(){
        if(!this.isStarted()){
            return;
        }
        this.started=false;
        this.worker.interrupt();
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Long getExpireTime() {
        return expireTime;
    }

    public void setExpireTime(Long expireTime) {
        this.expireTime = expireTime;
    }
    @Override
    public void addListener(QueueListener listener){
        logger.info("com.winit.core.mq.queue.impl.AsyncMessageQueue.addListener name="+this.name);
        this.listeners.add(listener);
    }
    @Override
    public void removeListener(Object object){
        if(this.listeners.size() == 0){
            return;
        }
        int i=listeners.size()-1;
        for(;i>=0;i--){
            QueueListener listener=listeners.get(i);
            if(listener.equals(object)){
                listeners.remove(i);
            }
        }
    }

    /**
     * 队列的监听线程, 启动后开始尝试从队列获取消息.
     *
     * 获取到消息后,就
     */
    class Worker extends Thread {
        Worker() {
        }

        public void run() {
            AsyncMessageQueue parent = AsyncMessageQueue.this;
            //AppenderAttachableImpl aai = parent.aai;

            //1. 如果线程是启动的,则进入这个死循环不再退出.
            while(parent.isStarted()) {
                try {
                    //1.1 尝试取一条消息(没有的时候就blocking)
                    Message message = parent.queue.take();
                    for(QueueListener listener: listeners){
                        //1.2 分发给所有的消费者.
                        listener.onMessage(message);
                    }
                } catch (InterruptedException var5) {

                    //9. 如果线程被意外终止\打断时, 就跳出这个死循环,进入下一步处理.
                    break;
                }
            }

            //2. 当"队列单例对象"是非启动状态(异常状态), 或者在费阶段出现了任何异常, 才会进入这个环节.
            cleanQueue(parent);
        }

        /**
         *  尝试将(异常停止掉的)队列中剩余的所有消息都取出来, 分发给现存的listener,并删除这些消息.
         * @param parent
         */
        private void cleanQueue(AsyncMessageQueue parent) {
            Iterator<Message> iterator = parent.queue.iterator();
            while(iterator.hasNext()) {
                Message message = iterator.next();
                for(QueueListener listener: listeners){
                    listener.onMessage(message);
                }
                parent.queue.remove(message);
            }
        }
    }
}
